/*
 * Copyright 2008-2009 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.hasor.neta.channel;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import net.hasor.cobble.concurrent.ThreadUtils;
import net.hasor.cobble.concurrent.future.Future;
import net.hasor.cobble.concurrent.timer.HashedWheelTimer;
import net.hasor.cobble.concurrent.timer.TimerTask;
import net.hasor.cobble.io.IOUtils;
import net.hasor.cobble.logging.Logger;
import net.hasor.neta.bytebuf.ByteBufAllocator;

/**
 * SoContext implements
 * @author 赵永春 (zyc@hasor.net)
 * @version : 2023-09-24
 */
public class SoContextService implements SoContext {
    private static final Logger                            logger    = Logger.getLogger(SoContextService.class);
    private final        AtomicLong                        nextID    = new AtomicLong(0);
    private final        NetConfig                         config;
    private final        NetManager                        manager;
    private final        ByteBufAllocator                  allocator;
    private final        ClassLoader                       useClassLoader;
    private final        SoThreadFactory                   useSoThreadFactory;
    //
    private final        List<Function<PlayLoad, Boolean>> listeners = new CopyOnWriteArrayList<>();
    //
    private final        HashedWheelTimer                  globalTimer;
    private final        ExecutorService                   ioExecutor;
    private final        SoEventExecutor                   eventExecutor;
    private final        ReentrantReadWriteLock            closeSyncLock;
    private final        Map<Long, SoChannel<?>>           channelMap;
    private final        Queue<NetChannel>                 channelList;
    private final        Queue<NetListen>                  listenList;
    private volatile     boolean                           closeStatus;

    SoContextService(NetConfig netConf, NetManager manager) {
        this.manager = manager;
        this.allocator = netConf.getBufAllocator() == null ? ByteBufAllocator.DEFAULT : netConf.getBufAllocator();
        this.config = Objects.requireNonNull(netConf);
        this.useClassLoader = this.config.getClassLoader() == null ? SoContextService.class.getClassLoader() : this.config.getClassLoader();

        if (netConf.getThreadFactory() == null) {
            this.useSoThreadFactory = (loader, nameTemplate) -> ThreadUtils.threadFactory(loader, nameTemplate, true);
        } else {
            this.useSoThreadFactory = netConf.getThreadFactory();
        }

        // timer
        ThreadFactory timerThread = ThreadUtils.daemonThreadFactory(this.useClassLoader, "Neta-Timer");
        this.globalTimer = new HashedWheelTimer(timerThread, 50, TimeUnit.MILLISECONDS);
        this.globalTimer.start();

        // io exec
        int defaultProcess = this.config.getIoThreads();
        if (defaultProcess < 1) {
            defaultProcess = Math.max(Runtime.getRuntime().availableProcessors() / 4, 1);
        }
        ThreadFactory ioThreadFactory = this.useSoThreadFactory.newFactory(this.useClassLoader, "Neta-IO-%s");
        this.ioExecutor = Executors.newFixedThreadPool(defaultProcess, ioThreadFactory);

        // task exec
        int taskWorkSize = netConf.getTaskThreads();
        if (taskWorkSize < 1) {
            taskWorkSize = Runtime.getRuntime().availableProcessors();
        }
        this.eventExecutor = new SoEventExecutor(this.useClassLoader, this.useSoThreadFactory, taskWorkSize, this.globalTimer);

        //
        this.closeStatus = false;
        this.closeSyncLock = new ReentrantReadWriteLock(true);
        this.channelMap = new ConcurrentHashMap<>();
        this.channelList = new ConcurrentLinkedQueue<>();
        this.listenList = new ConcurrentLinkedQueue<>();
    }

    public long nextID() {
        return nextID.incrementAndGet();
    }

    @Override
    public NetConfig getConfig() {
        return this.config;
    }

    @Override
    public ByteBufAllocator getByteBufAllocator() {
        return this.allocator;
    }

    @Override
    public SocketAddress getRemoteAddress(long channelId) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            return null;
        } else {
            return channel.getRemoteAddr();
        }
    }

    public ExecutorService getIoExecutor() {
        return this.ioExecutor;
    }

    /** Whether to accept link socket. */
    public boolean acceptChannel(SocketAddress remoteAddress) {
        if (this.closeStatus) {
            return false;
        }

        return true;
    }

    /** test the channel has been closed */
    @Override
    public boolean isClose(long channelId) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        return channel == null || channel.isClose();
    }

    /** close status. */
    public boolean isClose() {
        return this.closeStatus;
    }

    @Override
    public SoChannel<?> findChannel(long channelId) {
        return this.channelMap.get(channelId);
    }

    public void initChannel(SoChannel<?> channel, boolean init) throws Throwable {
        long channelId = channel.getChannelId();
        if (this.channelMap.containsKey(channelId)) {
            throw new SoException("channelId already exists.");
        }

        // add channel
        try {
            this.closeSyncLock.readLock().lock();

            this.channelMap.put(channel.getChannelId(), channel);
            if (channel.isListen()) {
                this.listenList.add((NetListen) channel);
            } else {
                this.channelList.add((NetChannel) channel);
            }
        } finally {
            this.closeSyncLock.readLock().unlock();
        }

        // init
        if (init && channel instanceof NetChannel) {
            NetChannel netChannel = (NetChannel) channel;
            ProtoStack<Object> protoStack = netChannel.protoStack;
            ProtoContextService protoCtx = netChannel.protoCtx;

            protoStack.onInit(protoCtx);
            if (!channel.isClose()) {
                protoStack.onActive(protoCtx);
            }

            if (!channel.isClose() && netChannel.getListen() != null) {
                netChannel.getListen().notifyAccept(netChannel);
            }
        }
    }

    @Override
    public SubscribeHolder subscribe(long channelId, PlayLoadListener listener) {
        return this.subscribe(p -> p.getSource().getChannelId() == channelId, listener);
    }

    @Override
    public SubscribeHolder subscribe(final Predicate<PlayLoad> select, final PlayLoadListener listener) {
        if (listener != null) {
            final Function<PlayLoad, Boolean> func = data -> {
                if (select == null || select.test(data)) {
                    listener.onEvent(data);
                    return true;
                } else {
                    return false;
                }
            };
            this.listeners.add(func);
            return () -> this.listeners.remove(func);
        } else {
            return null;
        }
    }

    /** trigger event */
    @Deprecated
    public void trigger(PlayLoad data) {
        String prefix;
        if (data.isInbound()) {
            prefix = "rcv";
        } else if (data.isOutbound()) {
            prefix = "snd";
        } else {
            prefix = "event";
        }

        boolean hasProcessed = false;
        for (Function<PlayLoad, Boolean> listener : this.listeners) {
            try {
                hasProcessed = hasProcessed | listener.apply(data);
            } catch (Exception e) {
                logger.error(prefix + "(" + data.getSource().getChannelId() + ") trigger " + listener.getClass().getName() + " has error " + e.getMessage(), e);
            }
        }

        if (!hasProcessed) {
            String msg = prefix + "(" + data.getSource().getChannelId() + ") There are no program at the tail of the ProtoStack, Skipping event: ";
            logger.warn(msg + data.getData());
        }
    }

    @Override
    public NetManager getNetManager() {
        return this.manager;
    }

    void foreachListen(Consumer<NetListen> consumer) {
        this.listenList.forEach(consumer);
    }

    /** close all socket, The method {@link #initChannel(SoChannel, boolean)} and {@link #closeAll(boolean)} are mutually exclusive */
    public void closeAll(boolean now) {
        // mark close is true.
        try {
            this.closeSyncLock.writeLock().lock();
            this.closeStatus = true;
        } finally {
            this.closeSyncLock.writeLock().unlock();
        }

        List<Future<?>> waitFinish = new LinkedList<>();

        // close all NetListen
        while (!this.listenList.isEmpty()) {
            NetListen listen = this.listenList.poll();
            if (listen != null) {
                if (now) {
                    listen.closeNow();
                } else {
                    waitFinish.add(listen.close());
                }
            }
        }

        // close all NetChannel
        while (!this.channelList.isEmpty()) {
            NetChannel channel = this.channelList.poll();
            if (channel != null) {
                if (now) {
                    channel.closeNow();
                } else {
                    waitFinish.add(channel.close());
                }
            }
        }

        // wait all finish
        while (true) {
            boolean allFinish = true;

            for (Future<?> future : waitFinish) {
                allFinish = future.isDone();
                if (!allFinish) {
                    break;
                }
            }

            if (!allFinish) {
                ThreadUtils.sleep(300);
            } else {
                break;
            }
        }
    }

    /** asynchronously copy data from swap to rcv/snd */
    public <T> Future<T> submitSoTask(DefaultSoTask task, T result) {
        return this.eventExecutor.submitSoTask(task, result);
    }

    /** Set up a timer */
    protected void newTimeout(TimerTask task, long delay, TimeUnit unit) {
        this.globalTimer.newTimeout(task, delay, unit);
    }

    public void notifyRcvUserEvent(long channelId, String stackName, SoUserEvent event) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("notifyRcvUserEvent, channel not found. channelId : " + channelId);
            return;
        }

        if (!(channel instanceof NetChannel)) {
            logger.error("only NetChannel can notifyRcvUserEvent. channelId : " + channelId);
            return;
        }

        try {
            NetChannel netChannel = (NetChannel) channel;
            netChannel.protoStack.onRcvUserEvent(netChannel.protoCtx, stackName, event);
        } catch (Throwable e) {
            SoException ee = e instanceof SoException ? (SoException) e : new SoRcvException(e.getMessage(), e);
            this.notifyRcvChannelException(channelId, false, ee);
        }
    }

    public void notifySndUserEvent(long channelId, String stackName, SoUserEvent event) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("notifySndUserEvent, channel not found. channelId : " + channelId);
            return;
        }

        if (!(channel instanceof NetChannel)) {
            logger.error("only NetChannel can notifySndUserEvent. channelId : " + channelId);
            return;
        }

        try {
            NetChannel netChannel = (NetChannel) channel;
            netChannel.protoStack.onSndUserEvent(netChannel.protoCtx, stackName, event);
        } catch (Throwable e) {
            SoException ee = e instanceof SoException ? (SoException) e : new SoSndException(e.getMessage(), e);
            this.notifySndChannelException(channelId, false, ee);
        }
    }

    public void notifyBindChannelException(long channelId, SoBindException e) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("channel not found. channelId : " + channelId, e);
            return;
        }

        if (channel instanceof NetListen) {
            logger.error("ERROR: bindFailed, " + e.getMessage(), e);
            this.doCloseChannel(channel, "bindFailed, " + e.getMessage(), e);
        } else {
            logger.error("only NetChannel can notifyBindException. channelId : " + channelId);
        }
    }

    public void notifyConnectChannelException(long channelId, boolean doClose, SoConnectException e) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("channel not found. channelId : " + channelId, e);
            return;
        }

        if (channel instanceof NetChannel) {
            this.doNotifyError(true, doClose, e, channel);
        } else {
            logger.error("only NetChannel can notifyConnectException. channelId : " + channelId);
        }
    }

    /** receiving new data */
    public void notifyRcvChannelData(long channelId, Object... rcvData) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("notifyRcvFailed, channel not found. channelId : " + channelId);
            return;
        }

        if (!(channel instanceof NetChannel)) {
            logger.error("only NetChannel can notifyData. channelId : " + channelId);
            return;
        }

        try {
            ((NetChannel) channel).notifyRcv(rcvData);
        } catch (Throwable e) {
            SoException ee = e instanceof SoException ? (SoException) e : new SoRcvException(e.getMessage(), e);
            this.notifyRcvChannelException(channelId, true, ee);
        }
    }

    public void notifyRcvChannelException(long channelId, boolean doClose, SoException e) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("channel not found. channelId : " + channelId, e);
            return;
        }

        if (channel instanceof NetChannel) {
            this.doNotifyError(true, doClose, e, channel);
        } else {
            logger.error("only NetChannel can notifyRcvException. channelId : " + channelId);
        }
    }

    public void notifySndChannelException(long channelId, boolean doClose, SoException e) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel == null) {
            logger.error("channel not found. channelId : " + channelId, e);
            return;
        }

        if (channel instanceof NetChannel) {
            this.doNotifyError(false, doClose, e, channel);
        } else {
            logger.error("only NetChannel can notifySendException. channelId : " + channelId);
        }
    }

    private void doNotifyError(boolean isRcv, boolean doClose, SoException e, SoChannel<?> channel) {
        try {
            ((NetChannel) channel).notifyError(isRcv, e);
            if (doClose) {
                this.doCloseChannel(channel, "close channel for exception " + e.getMessage(), e);
            }
        } catch (Throwable ee) {
            String errorMsg = "close channel for unhandled exception " + ee.getMessage();
            logger.error(errorMsg, ee);
            this.doCloseChannel(channel, errorMsg, ee);
        }
    }

    public void notifyChannelClose(long channelId, boolean remote) {
        SoChannel<?> channel = this.channelMap.get(channelId);
        if (channel != null) {
            this.doCloseChannel(channel, "closed from " + (remote ? "remote" : "local"), null);
        }
    }

    private void doCloseChannel(SoChannel<?> channel, String message, Throwable e) {
        if (e == null) {
            logger.info("channel(" + channel.getChannelId() + ") " + message);
        } else {
            logger.error(message, e);
        }

        // clean wQueue
        if (channel instanceof NetChannel) {
            NetChannel netChannel = (NetChannel) channel;
            IOUtils.closeQuietly(netChannel.asyncChannel);
            netChannel.closeStatus.set(true);

            // purge data
            netChannel.wContext.purge(e);

            // on close event
            try {
                netChannel.protoStack.onClose(netChannel.protoCtx);
            } catch (Exception ignore) {
                //
            } finally {
                NetListen forListen = netChannel.getListen();
                if (forListen != null) {
                    forListen.notifyClose(netChannel);
                }

                this.channelMap.remove(channel.getChannelId());
            }

            try {
                netChannel.closeFuture.completed(netChannel);
            } catch (Exception ignore) {
                //
            }
        } else {
            NetListen netListen = (NetListen) channel;
            IOUtils.closeQuietly(netListen.channel);
            netListen.closeStatus.set(true);

            this.listenList.remove(channel);
            logger.info("listen(" + channel.getChannelId() + ") closed, port :" + netListen.getListenPort());
        }
    }
}